// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: chunhui

#include "cc/appcoin2/support/feed/interval_feed_tool/interval_feed_consumer.h"

#include <algorithm>
#include <limits>

namespace interval_feed {

void IntervalFeedConsumer::InitTimestamp(int resolution) {
  all_ts_.resize(n_intervals_);
  int64_t NS = 1'000'000'000LL;
  for (int i = 0; i < n_intervals_; i++) {
    all_ts_[i] = resolution * NS * (i + 1) + start_ts_;
  }
}

int64_t IntervalFeedConsumer::GetIntervalId(int64_t ts) const {
  if (ts >= start_ts_) {
    return (ts - start_ts_) / (resolution_ * 1'000'000'000LL);
  }
  return -1;
}

void IntervalFeedConsumer::Consume(const FeedUpdate& upd, FeedType feed_type) {
  const std::string& key = upd.product().absolute_norm();
  if (symbol_id_.find(key) == symbol_id_.end()) {
    symbol_id_[key] = all_symbols_.size();
    previous_interval_id_[all_symbols_.size()] = 0;
    all_symbols_.push_back(key);
    const auto& product = upd.product();
    std::string full_symbol = fmt::format(
        "{}:{}:{}",
        MarketTypeToString(product.market()),
        ExchangeTypeToString(product.exchange()),
        product.relative_norm());
    all_products_.push_back(CreateProductFromUniqueString(full_symbol, upd.timestamp()));
    symbol_to_calculators_.emplace_back(n_intervals_);
  }
  int symbol_id = symbol_id_[key];

  int64_t ts = upd.timestamp();
  int64_t interval_id = GetIntervalId(ts);
  if (interval_id < 0 || interval_id >= n_intervals_) {
    return;
  }
  auto& calculator = symbol_to_calculators_[symbol_id][interval_id];

  if (upd.is_book() && feed_type == FeedType::BOOK_FEED) {
    const IBook& book = upd.book();
    if (book.Bid0() && book.Ask0()) {
      calculator.UpdateBook(ts, book.Bid0()->price, book.Ask0()->price);
      calculator.UpdateTimestamp(
          TsType::BOOK_FTET,
          (book.Timestamp() == 0) ? 0 : ts - book.Timestamp());
    } else if (upd.mea().String() == "Options.Deribit.v2") {
      calculator.UpdateBook(ts, std::numeric_limits<double>::quiet_NaN(), std::numeric_limits<double>::quiet_NaN());
      calculator.UpdateTimestamp(
          TsType::BOOK_FTET,
          (book.Timestamp() == 0) ? 0 : ts - book.Timestamp());
    }
    if(upd.mea().String() == "Options.Deribit.v2" && interval_id - previous_interval_id_[symbol_id] > 1) {
      auto p_ohlc =  symbol_to_calculators_[symbol_id][previous_interval_id_[symbol_id]].ohlc_;
      for (auto it =
               symbol_to_calculators_[symbol_id].begin() + previous_interval_id_[symbol_id] + 1;
           it != symbol_to_calculators_[symbol_id].begin() + interval_id;
           it++) {
        it->ohlc_[PriceType::BID0] = p_ohlc[PriceType::BID0];
        it->ohlc_[PriceType::ASK0] = p_ohlc[PriceType::ASK0];
        it->ohlc_[PriceType::MID] = p_ohlc[PriceType::MID];
      }
    }
    previous_interval_id_[symbol_id] = interval_id;
  } else if (upd.is_trade() && feed_type == FeedType::TRADE_FEED) {
    const Trade& trade = upd.trade();
    calculator.UpdateTrade(trade.price, trade.fill_qty, trade.side);
    calculator.UpdateTimestamp(
        TsType::TRADE_FTET,
        (trade.timestamp == 0) ? 0 : ts - trade.timestamp);
  } else if (feed_type == FeedType::LIQUIDATION_FEED) {
    const LiquidationOrder& order = upd.liquidation_order();
    if (order.timestamp > 0) {
      calculator.UpdateLiquidation(order.price, order.qty, order.sign());
    }
  } else if (feed_type == FeedType::OPEN_INTEREST_FEED) {
    const OpenInterest& open_interest = upd.open_interest();
    if (open_interest.timestamp > 0) {
      calculator.UpdateGeneralOhlc(PriceType::OPEN_INTEREST, open_interest.open_interest_qty);
    }
  } else if (feed_type == FeedType::NAV_FEED) {
    const Nav& nav = upd.nav();
    if (nav.timestamp > 0) {
      calculator.UpdateGeneralOhlc(PriceType::NAV, nav.nav);
    }
  } else if (feed_type == FeedType::INDEX_FEED) {
    const auto& index = upd.index();
    if (index.timestamp > 0) {
      calculator.UpdateGeneralOhlc(PriceType::INDEX, index.price);
    }
  } else if (feed_type == FeedType::MARK_PRICE_FEED) {
    const MarkPrice& mark_price = upd.mark_price();
    if (mark_price.timestamp > 0) {
      calculator.UpdateGeneralOhlc(PriceType::MARK_PRICE, mark_price.mark_price);
    }
  } else if (feed_type == FeedType::FUNDING_RATE_FEED) {
    const FundingRate& funding_rate = upd.funding_rate();
    if (funding_rate.timestamp > 0) {
      calculator.UpdateTimestamp(TsType::FUNDING_TIME, funding_rate.funding_time);
      calculator.UpdateGeneralOhlc(PriceType::FUNDING_RATE, funding_rate.funding_rate);
    }
  } else if (feed_type == FeedType::MARKET_ANALYSIS_FEED) {
    const MarketAnalysis& ma = upd.market_analysis();
    calculator.UpdateMarketAnalysis(ma.total_supply,
                                    ma.max_supply,
                                    ma.circulating_supply,
                                    ma.market_cap_in_usd,
                                    ma.tvl_in_usd);
  } else if (feed_type == FeedType::LONG_SHORT_RATIO) {
    if (upd.is_top_long_short_account_ratio()) {
      const TopLongShortAccountRatio* lsr = upd.top_long_short_account_ratio().Get().value();
      calculator.UpdateLongShortRatio(
          lsr->long_short_ratio,
          LongShortRatioType::TOP_LONG_SHORT_ACCOUNT_RATIO);
    } else if (upd.is_top_long_short_position_ratio()) {
      const TopLongShortPositionRatio* lsr = upd.top_long_short_position_ratio().Get().value();
      calculator.UpdateLongShortRatio(
          lsr->long_short_ratio,
          LongShortRatioType::TOP_LONG_SHORT_POSITION_RATIO);
    } else {
      const GlobalLongShortAccountRatio* lsr = upd.global_long_short_account_ratio().Get().value();
      calculator.UpdateLongShortRatio(
          lsr->long_short_ratio,
          LongShortRatioType::GLOBAL_LONG_SHORT_ACCOUNT_RATIO);
    }
  }
}

void IntervalFeedConsumer::Write(const std::string& file_prefix, const WriterOptions& option) {
  sort(all_symbols_.begin(), all_symbols_.end());
  if (option.enable_ohlc) {
    WriteOhlc(file_prefix);
  }
  if (option.enable_vwap) {
    WriteVwap(file_prefix);
  }
  if (option.enable_volume) {
    WriteVolume(file_prefix);
  }
  if (option.enable_other_stats) {
    WriteOtherStats(file_prefix);
  }
  if (option.enable_time) {
    WriteTime(file_prefix);
  }
  if (option.enable_market_analysis) {
    WriteMarketAnalysis(file_prefix);
  }
}

void IntervalFeedConsumer::PopulateData(
    const std::string& data_frame,
    OhlcType ohlc_type,
    PriceType price_type,
    VwapType vwap_type,
    VolumeType volume_type,
    QuoteType quote_type,
    OtherStatsType other_stats_type,
    TsType ts_type,
    MarketAnalysisType ma_type,
    LongShortRatioType lsr_type,
    std::vector<double>* data) {
  for (int interval_id = 0; interval_id < n_intervals_; interval_id++) {
    for (const auto& symbol : all_symbols_) {
      int symbol_id = symbol_id_[symbol];
      const auto& calculator = symbol_to_calculators_[symbol_id][interval_id];
      if (data_frame == "OHLC") {
        data->push_back(calculator.ReadOhlc(price_type, ohlc_type));
      } else if (data_frame == "VWAP") {
        data->push_back(calculator.ReadVwap(vwap_type));
      } else if (data_frame == "VOLUME") {
        if (quote_type == QuoteType::DOLLAR) {
          if (all_products_[symbol_id]->market() == MarketType::Options && all_products_[symbol_id]->exchange() == ExchangeType::Deribit){
            data->push_back(calculator.ReadDeribitOptionDollarVolume(volume_type));
          } else {
            const auto& pi = product_cache_->holder(*all_products_[symbol_id]).product_info();
            data->push_back(calculator.ReadDollarVolume(volume_type, pi));
          }
        } else {
          data->push_back(calculator.ReadVolume(volume_type));
        }
      } else if (data_frame == "TIME") {
        data->push_back(calculator.ReadOhlc(ts_type, ohlc_type));
      } else if (data_frame == "OTHERS") {
        if (other_stats_type == OtherStatsType::STD) {
          data->push_back(calculator.ReadStd());
        } else if (other_stats_type == OtherStatsType::SPREAD) {
          data->push_back(calculator.ReadTimeWeightedSpread(all_ts_[interval_id]));
        } else if (lsr_type == LongShortRatioType::TOP_LONG_SHORT_ACCOUNT_RATIO) {
          data->push_back(calculator.ReadLongShortRatio(lsr_type));
        } else if (lsr_type == LongShortRatioType::TOP_LONG_SHORT_POSITION_RATIO) {
          data->push_back(calculator.ReadLongShortRatio(lsr_type));
        } else if (lsr_type == LongShortRatioType::GLOBAL_LONG_SHORT_ACCOUNT_RATIO) {
          data->push_back(calculator.ReadLongShortRatio(lsr_type));
        } else {
            data->push_back(std::numeric_limits<double>::quiet_NaN());
          }
        } else if (data_frame == "MARKET_ANALYSIS") {
          data->push_back(calculator.ReadMarketAnalysis(ma_type));
        } else {
          data->push_back(std::numeric_limits<double>::quiet_NaN());
        }
    }
  }
}

void IntervalFeedConsumer::WriteOhlc(const std::string& file_prefix) {
  PriceType price_types[] = {
      PriceType::BID0,
      PriceType::ASK0,
      PriceType::MID,
      PriceType::TRADE,
      PriceType::OPEN_INTEREST,
      PriceType::NAV,
      PriceType::INDEX,
      PriceType::MARK_PRICE,
      PriceType::FUNDING_RATE};
  OhlcType ohlc_types[] = {OhlcType::OPEN, OhlcType::CLOSE, OhlcType::HIGH, OhlcType::LOW};

  std::string filename = "ohlc";
  std::string filepath = file_prefix + "--" + filename + ".h5";

  std::vector<double> data;
  int width = all_symbols_.size();
  int height = n_intervals_;
  data.reserve(all_symbols_.size() * n_intervals_);
  TimedDataUniverseFixedDumper writer(filepath, width, height);
  writer.set_universe(all_symbols_);
  writer.set_timestamp(all_ts_);
  for (const auto& price_type : price_types) {
    for (const auto& ohlc_type : ohlc_types) {
      data.clear();
      PopulateData(
          "OHLC",
          ohlc_type,
          price_type,
          VwapType::UNKNOWN_VWAP_TYPE,
          VolumeType::UNKNOWN_VOLUME_TYPE,
          QuoteType::UNKNOWN_QUOTE_TYPE,
          OtherStatsType::UNKNOWN_OTHER_STATS_TYPE,
          TsType::UNKNOWN_TS_TYPE,
          MarketAnalysisType::UNKNOWN_MARKET_ANALYSIS_TYPE,
          LongShortRatioType::UNKNOWN_LONG_SHORT_RATIO_TYPE,
          &data);
      std::string column_name = OhlcType_Name(ohlc_type) + "_" + PriceType_Name(price_type);
      writer.set_column_double(column_name, data);
    }
  }
}

void IntervalFeedConsumer::WriteVwap(const std::string& file_prefix) {
  VwapType vwap_types[] = {VwapType::VWAP_BUY, VwapType::VWAP_SELL, VwapType::VWAP};

  std::string filename = "vwap";
  std::string filepath = file_prefix + "--" + filename + ".h5";

  std::vector<double> data;
  int width = all_symbols_.size();
  int height = n_intervals_;
  data.reserve(all_symbols_.size() * n_intervals_);
  TimedDataUniverseFixedDumper writer(filepath, width, height);
  writer.set_universe(all_symbols_);
  writer.set_timestamp(all_ts_);
  for (const auto& vwap_type : vwap_types) {
    data.clear();
    PopulateData(
        "VWAP",
        OhlcType::UNKNOWN_OHLC_TYPE,
        PriceType::UNKNOWN_PRICE_TYPE,
        vwap_type,
        VolumeType::UNKNOWN_VOLUME_TYPE,
        QuoteType::UNKNOWN_QUOTE_TYPE,
        OtherStatsType::UNKNOWN_OTHER_STATS_TYPE,
        TsType::UNKNOWN_TS_TYPE,
        MarketAnalysisType::UNKNOWN_MARKET_ANALYSIS_TYPE,
        LongShortRatioType::UNKNOWN_LONG_SHORT_RATIO_TYPE,
        &data);
    writer.set_column_double(VwapType_Name(vwap_type), data);
  }
}

void IntervalFeedConsumer::WriteVolume(const std::string& file_prefix) {
  VolumeType volume_types[] = {
      VolumeType::VOLUME_BUY,
      VolumeType::VOLUME_SELL,
      VolumeType::VOLUME,
      VolumeType::VOLUME_LIQUIDATION_BUY,
      VolumeType::VOLUME_LIQUIDATION_SELL};

  QuoteType quote_types[] = {QuoteType::COIN, QuoteType::DOLLAR};
  std::string filename = "volume";
  std::string filepath = file_prefix + "--" + filename + ".h5";

  std::vector<double> data;
  int width = all_symbols_.size();
  int height = n_intervals_;
  data.reserve(all_symbols_.size() * n_intervals_);
  TimedDataUniverseFixedDumper writer(filepath, width, height);
  writer.set_universe(all_symbols_);
  writer.set_timestamp(all_ts_);
  for (const auto& quote_type : quote_types) {
    for (const auto& volume_type : volume_types) {
      data.clear();
      PopulateData(
          "VOLUME",
          OhlcType::UNKNOWN_OHLC_TYPE,
          PriceType::UNKNOWN_PRICE_TYPE,
          VwapType::UNKNOWN_VWAP_TYPE,
          volume_type,
          quote_type,
          OtherStatsType::UNKNOWN_OTHER_STATS_TYPE,
          TsType::UNKNOWN_TS_TYPE,
          MarketAnalysisType::UNKNOWN_MARKET_ANALYSIS_TYPE,
          LongShortRatioType::UNKNOWN_LONG_SHORT_RATIO_TYPE,
          &data);
      std::string quote_type_name = "_" + QuoteType_Name(quote_type);
      if (quote_type == QuoteType::COIN) {
        quote_type_name = "";  // For back compatibility
      }
      writer.set_column_double(VolumeType_Name(volume_type) + quote_type_name, data);
    }
  }
}

void IntervalFeedConsumer::WriteOtherStats(const std::string& file_prefix) {
  OtherStatsType other_types[] = {OtherStatsType::STD, OtherStatsType::SPREAD};

  std::string filename = "other_stats";
  std::string filepath = file_prefix + "--" + filename + ".h5";

  std::vector<double> data;
  int width = all_symbols_.size();
  int height = n_intervals_;
  data.reserve(all_symbols_.size() * n_intervals_);
  TimedDataUniverseFixedDumper writer(filepath, width, height);
  writer.set_universe(all_symbols_);
  writer.set_timestamp(all_ts_);
  for (const auto& other_type : other_types) {
    data.clear();
    PopulateData(
        "OTHERS",
        OhlcType::UNKNOWN_OHLC_TYPE,
        PriceType::UNKNOWN_PRICE_TYPE,
        VwapType::UNKNOWN_VWAP_TYPE,
        VolumeType::UNKNOWN_VOLUME_TYPE,
        QuoteType::UNKNOWN_QUOTE_TYPE,
        other_type,
        TsType::UNKNOWN_TS_TYPE,
        MarketAnalysisType::UNKNOWN_MARKET_ANALYSIS_TYPE,
        LongShortRatioType::UNKNOWN_LONG_SHORT_RATIO_TYPE,
        &data);
    writer.set_column_double(OtherStatsType_Name(other_type), data);
  }

  LongShortRatioType lsr_types[] = {
    LongShortRatioType::GLOBAL_LONG_SHORT_ACCOUNT_RATIO,
    LongShortRatioType::TOP_LONG_SHORT_ACCOUNT_RATIO,
    LongShortRatioType::TOP_LONG_SHORT_POSITION_RATIO
  };
  for (const auto& lsr_type : lsr_types) {
    data.clear();
    PopulateData(
        "OTHERS",
        OhlcType::UNKNOWN_OHLC_TYPE,
        PriceType::UNKNOWN_PRICE_TYPE,
        VwapType::UNKNOWN_VWAP_TYPE,
        VolumeType::UNKNOWN_VOLUME_TYPE,
        QuoteType::UNKNOWN_QUOTE_TYPE,
        OtherStatsType::UNKNOWN_OTHER_STATS_TYPE,
        TsType::UNKNOWN_TS_TYPE,
        MarketAnalysisType::UNKNOWN_MARKET_ANALYSIS_TYPE,
        lsr_type,
        &data);
    writer.set_column_double(LongShortRatioType_Name(lsr_type), data);
  }
}

void IntervalFeedConsumer::WriteTime(const std::string& file_prefix) {
  std::vector<std::pair<std::vector<TsType>, std::vector<OhlcType>>>
      ts_ohlc_set = {
          {{TsType::FUNDING_TIME}, {OhlcType::OPEN, OhlcType::CLOSE}},
          {{TsType::BOOK_FTET, TsType::TRADE_FTET}, {OhlcType::MEAN}}};

  std::string filename = "time";
  std::string filepath = file_prefix + "--" + filename + ".h5";

  std::vector<double> data;
  int width = all_symbols_.size();
  int height = n_intervals_;
  data.reserve(all_symbols_.size() * n_intervals_);
  TimedDataUniverseFixedDumper writer(filepath, width, height);
  writer.set_universe(all_symbols_);
  writer.set_timestamp(all_ts_);
  for (const auto& [ts_types, ohlc_types] : ts_ohlc_set) {
    for (const auto& ts_type : ts_types) {
      for (const auto& ohlc_type : ohlc_types) {
        data.clear();
        PopulateData(
            "TIME",
            ohlc_type,
            PriceType::UNKNOWN_PRICE_TYPE,
            VwapType::UNKNOWN_VWAP_TYPE,
            VolumeType::UNKNOWN_VOLUME_TYPE,
            QuoteType::UNKNOWN_QUOTE_TYPE,
            OtherStatsType::UNKNOWN_OTHER_STATS_TYPE,
            ts_type,
            MarketAnalysisType::UNKNOWN_MARKET_ANALYSIS_TYPE,
            LongShortRatioType::UNKNOWN_LONG_SHORT_RATIO_TYPE,
            &data);
        std::string column_name = OhlcType_Name(ohlc_type) + "_" + TsType_Name(ts_type);
        writer.set_column<double>(column_name, data);
      }
    }
  }
}

void IntervalFeedConsumer::WriteMarketAnalysis(const std::string& file_prefix) {
  MarketAnalysisType ma_types[] = {
    MarketAnalysisType::TOTAL_SUPPLY,
    MarketAnalysisType::MAX_SUPPLY,
    MarketAnalysisType::CIRCULATING_SUPPLY,
    MarketAnalysisType::MARKET_CAP_IN_USD,
    MarketAnalysisType::TVL_IN_USD
  };

  std::string filename = "market_analysis";
  std::string filepath = file_prefix + "--" + filename + ".h5";

  std::vector<double> data;
  int width = all_symbols_.size();
  int height = n_intervals_;
  data.reserve(all_symbols_.size() * n_intervals_);
  TimedDataUniverseFixedDumper writer(filepath, width, height);
  writer.set_universe(all_symbols_);
  writer.set_timestamp(all_ts_);
  for (const auto& ma_type : ma_types) {
    data.clear();
    PopulateData(
      "MARKET_ANALYSIS",
      OhlcType::UNKNOWN_OHLC_TYPE,
      PriceType::UNKNOWN_PRICE_TYPE,
      VwapType::UNKNOWN_VWAP_TYPE,
      VolumeType::UNKNOWN_VOLUME_TYPE,
      QuoteType::UNKNOWN_QUOTE_TYPE,
      OtherStatsType::UNKNOWN_OTHER_STATS_TYPE,
      TsType::UNKNOWN_TS_TYPE,
      ma_type,
      LongShortRatioType::UNKNOWN_LONG_SHORT_RATIO_TYPE,
      &data);
    writer.set_column_double(MarketAnalysisType_Name(ma_type), data);
  }
}

}  // namespace interval_feed
